# -*- coding: utf-8 -*-
"""
Created on Fri May  9 15:41:44 2025

@author: tianr
"""

import os
import pandas as pd
import gc
import time
import csv
import json


#读取数据集文件夹中所有的.parquet数据文件
def read_parquet_folder(folder_path):
    all_data = []
    # 遍历文件夹中的所有文件
    for filename in os.listdir(folder_path):
        if filename.endswith('.parquet'):
            file_path = os.path.join(folder_path, filename)
            try:
                # 读取和分析目标相关的指定列
                columns_to_read = ['purchase_history']
                df = pd.read_parquet(file_path, columns=columns_to_read)
                all_data.append(df)
            except Exception as e:
                print(f"读取文件 {file_path} 时出错: {e}")

    # 合并所有数据
    combined_df = pd.concat(all_data, ignore_index=True)
    #垃圾回收多余的内存
    del all_data
    gc.collect()
    return combined_df

"""
从CSV文件中读取JSON列并提取指定字段保存为新的CSV
参数:
    csv_file_path: 输入CSV文件路径（包含JSON字符串列）
    output_file: 输出CSV文件路径
"""     
def extract_to_csv(csv_file_path, output_file):
    extracted_data = []
    try:
        # 读取CSV文件
        df = pd.read_csv(csv_file_path)
        
        # 假设CSV的第一列包含JSON字符串
        json_column = df.columns[0]
        
        # 遍历每一行
        for _, row in df.iterrows():
            json_str = row[json_column]
            try:
                # 解析JSON
                data = json.loads(json_str)
                
                # 提取基本字段
                payment_method = data.get('payment_method', '')
                payment_status = data.get('payment_status', '')
                purchase_date = data.get('purchase_date', '')
                
                # 提取items中的id列表
                items = data.get('items', [])
                item_ids = [item.get('id', '') for item in items]
                
                # 添加到结果列表
                extracted_data.append({
                    'items': item_ids,
                    'items_count': len(item_ids),
                    'payment_method': payment_method,
                    'payment_status': payment_status,
                    'purchase_date': purchase_date
                })
            except json.JSONDecodeError as e:
                print(f"警告：JSON解析错误 - 行 {_}: {e}")
            except Exception as e:
                print(f"警告：处理数据时出错 - 行 {_}: {e}")
    except FileNotFoundError:
        print(f"错误：文件 '{csv_file_path}' 不存在")
    except Exception as e:
        print(f"错误：读取CSV文件时出错 - {e}")
    
    # 转换为DataFrame并保存为CSV
    if extracted_data:
        result_df = pd.DataFrame(extracted_data)
        del extracted_data
        gc.collect()
        result_df.to_csv(
            output_file,
            index=False,
            encoding='utf-8-sig',
            quoting=csv.QUOTE_NONNUMERIC,
            escapechar='\\'
        )
        print(f"成功保存数据到 {output_file}")
        print(f"共处理 {len(extracted_data)} 条有效记录")
    else:
        print("警告：没有有效数据可保存")
        
def main(input_folder, output_csv):
    """主函数"""
    start_time = time.time()

    try:
        print("正在读取并处理数据...")
        data = read_parquet_folder(input_folder)
        df = pd.DataFrame(data)
        
        # 处理可能的乱码问题
        for col in df.columns:
            # 处理二进制列
            if df[col].dtype == object and isinstance(df[col].iloc[0], (bytes, bytearray)):
                df[col] = df[col].apply(
                    lambda x: x.decode('utf-8', errors='replace') if x else None
                )
            # 处理其他非字符串列
            elif not pd.api.types.is_string_dtype(df[col]):
                df[col] = df[col].astype(str)

        # 保存CSV（解决中文乱码）
        df.to_csv(
            output_csv,
            index=False,
            encoding='utf-8-sig',  # 添加BOM头（兼容Excel）
            quoting=csv.QUOTE_NONNUMERIC,  # 非数字内容加引号
            escapechar='\\',  # 转义特殊字符
        )

        # 打印统计信息
        end_time = time.time()
        print(f"\n转换完成！")
        print(f"- 实际提取行数: {len(df)}")
        print(f"- 耗时: {end_time - start_time:.2f} 秒")
    except Exception as e:
        print(f"程序运行出错: {str(e)}")
    finally:
        if 'df' in locals():
            del df
            gc.collect()

    elapsed_time = time.time() - start_time
    print(f"purchase history 记录已保存到: {output_csv}")
    print(f"总运行时间: {elapsed_time:.2f}秒")
    
    
#提取所有数据集中的purchase history的记录，并保存到本地purchase_history.csv文件中
if __name__ == "__main__":
    #指定数据集路径    
    input_folder1 = "D:/DTR/MyWork/2025/DataMining/personalProject/10G_data_new"
    input_folder2 = "D:/DTR/MyWork/2025/DataMining/personalProject/30G_data_new"
    output_csv1 = "D:/DTR/MyWork/2025/DataMining/personalProject/output/purchase_history_10G.csv"
    output_csv2 = "D:/DTR/MyWork/2025/DataMining/personalProject/output/purchase_history_30G.csv"

    main(input_folder1, output_csv1)
    main(input_folder2, output_csv2)
    
    output_file1 = "D:/DTR/MyWork/2025/DataMining/personalProject/output/items_10G_one.csv" 
    output_file2 = "D:/DTR/MyWork/2025/DataMining/personalProject/output/items_30G_one.csv"
    
    extract_to_csv(output_csv2, output_file2)
    
